/*
 * MasterServer.cpp
 *
 *  Created on: Feb 18, 2012
 *      Author: hduser
 */

#include "CloudClient.h"
#include "ExceptionHandler.h"
#include "DBHelper.h"
#include "CommonUtilities.h"
#include "MasterServer.h"
#include "CDNBitStreamExtract.h"

using namespace cloud;
using namespace utilities;
using namespace database;
using namespace logging;
using namespace sql;

const char* HDFS_CLOUD_SERVER_HOST = "localhost";
const int HDFS_CLOUD_SERVER_PORT = 54310;
const int STATUS_ERROR = 3;
const int STATUS_ACTIVE = 1;
const int STATUS_INACTIVE = 2;

MasterServer::MasterServer() {
	// TODO Auto-generated constructor stub

}

MasterServer::~MasterServer() {
	// TODO Auto-generated destructor stub
}

int MasterServer::ProcessFile(int VendorID, char* fileName) {

	try {
		DBHelper* dbChReq = new	 DBHelper();
		Logger* logger = new Logger();
		CommonUtilities* utilities = new CommonUtilities();
		CloudClient* client = new CloudClient((char*)HDFS_CLOUD_SERVER_HOST, HDFS_CLOUD_SERVER_PORT);
		hdfsFileInfo* fileInfo;
		time_t fiLastModified;
		size_t fiFileSize;

		bool bError = false;

		fileInfo = client->getFileInfoFromHDFS(fileName);
		fiLastModified = fileInfo->mLastMod;
		fiFileSize = fileInfo->mSize;

		// check this file in request table
		ResultSet* req = dbChReq->checkRequest(VendorID, fileName);
		if (req == NULL) {
			bError = true;
		} else {
			if (req->rowsCount() > 0) {
				req->first();
				int iOldRequestID = req->getInt("RequestID");
				SQLString sqls = req->getString("DateUpdated");
				char* OldDateUpdated = (char*)malloc(sqls.length());
				memcpy((void*)OldDateUpdated,(void*)sqls.c_str(),sqls.length());
				size_t OldSize = (size_t)req->getInt64("FileSize");
				dbChReq->disconnectDB();
				cout<<"iOldRequestID = "<<iOldRequestID<<" OldDateUpdated = "<<OldDateUpdated<<" OldSize = "<<OldSize<<endl;
				if (fileInfo != NULL) {
					if (utilities->isUpdated(fiLastModified, fiFileSize, OldDateUpdated, OldSize))
					{

						//BLOCK TO DELete inactive Request folders
						DBHelper* dbARI = new DBHelper();
						ResultSet* ari = dbARI->getActiveRequestItems(VendorID, fileName);

						if (ari == NULL)
						{
							bError = true;
						}
						else
						{
							if (ari->rowsCount() > 0)
							{
								cerr<<"Entered the ari block"<<endl;
								while(ari->next())
								{

									//cerr<<"chunkindex is :"<<ari->getInt("ChunkIndex")<<endl;
									//cerr<<"host name is :"<<(char*)ari->getString("HostName").c_str();
									//cerr<<"PORT name is :"<<(char*)ari->getString("ServicePort").c_str();

									// TODO
									// call method to delete chunk on edge server
									client->deleteFileFolderfromEdgeServer(VendorID, fileName,ari->getInt("ChunkIndex"), (char*)ari->getString("HostName").c_str(), ari->getInt("ServicePort"));

								}
							} else {
								bError = true;
							}
						}

						delete ari;

								//BLOCK TO DELete inactive Request folders
						DBHelper* dbMkReq = new DBHelper();
						dbMkReq->makeRequestInactive(iOldRequestID);
						dbMkReq->disconnectDB();
					} else {
						return 1;
					}
				} else {
					bError = true;
				}
			}
		}

		delete req;

		if (bError) {
			logger->logMessage("NULL", "NULL", "NULL", "MasterServer::ProcessFile", "Error in validating current Request queue", fileName);
			return -2; // db error
		}


		if (bError) {
			logger->logMessage("NULL", "NULL", "NULL", "MasterServer::ProcessFile", "Error while deleting old chunks", fileName);
			return -2; // db error
		}

		DBHelper* dbESG = new DBHelper();
		int iEdgeServerGroupID;
		ResultSet* esg = dbESG->getEdgeServerGroup();
		cout<<"here 1"<<endl;
		if (esg == NULL) {
			cout<<"here 2"<<endl;
			bError = true;
		} else {
			if (esg->first()) {
				cout<<"here 3"<<endl;
				iEdgeServerGroupID = esg->getInt("EdgeServerGroupID");
			} else {
				cout<<"here 4"<<endl;
				bError = true;
			}
		}

		delete esg;
		dbESG->disconnectDB();

		if (bError) {
			logger->logMessage("NULL", "NULL", "NULL", "MasterServer::ProcessFile", "Error in getting Edge Server Group", fileName);
			return -3; // es error
		}

		DBHelper* dbES = new DBHelper();
		ResultSet* esList = dbES->getEdgeServersInGroup(iEdgeServerGroupID);
		int iChunkCount = 0;
		int iRequestID;
		cout<<"here 5"<<endl;
		if (esList == NULL) {
			cout<<"error 1"<<endl;
			bError = true;
		} else {
			if (esList->rowsCount() > 0) {
				esList->first();
				iChunkCount = client->calculateNumberOfChunks(fileName);
				cout<<"iChunkCount = "<<iChunkCount<<endl;
				if (iChunkCount > 0) {
					cout<<"here 6"<<endl;
					char* stime = new char[22];
					utilities->getTimeString(fiLastModified, stime, 22);
					cout<<"fiLastModified = "<<fiLastModified<<endl;
					cout<<"fiFileSize = "<<fiFileSize<<endl;
					cout<<stime<<endl;
					DBHelper* dbCReq = new DBHelper();
					ResultSet* creq = dbCReq->createRequest(VendorID, fileName, stime, fiFileSize, iChunkCount, iEdgeServerGroupID);
					cout<<"here 7"<<endl;
					if (creq == NULL) {
						cout<<"error 2"<<endl;
						bError = true;
					} else {
						cout<<"here 8"<<endl;
						if (creq->rowsCount() > 0) {
							creq->first();
							iRequestID = creq->getInt("RequestID");
							cout<<"iRequestID ="<<iRequestID<<endl;
							dbCReq->disconnectDB();
							for(int iCtr = 1; iCtr <= iChunkCount; iCtr++) {
								try {
									/*
									struct chunkargs* args = (struct chunkargs*)malloc(sizeof(struct chunkargs));
									cout<<"host name = "<<(char*)esList->getString("HostName").c_str()<<endl;
									cout<<"host IP = "<<(char*)esList->getString("HostIP").c_str()<<endl;
									args->edgeserver_host = (char*)esList->getString("HostIP").c_str();
									args->edgeserver_port = esList->getInt("ServicePort");
									args->VendorID = VendorID;
									args->hdfsfileName = fileName;
									args->chunkIndex = iCtr;
									args->totalChunks = iChunkCount;
									*/
									int iRet = client->processChunk(VendorID, fileName, iCtr, iChunkCount, (char*)esList->getString("HostName").c_str(), esList->getInt("ServicePort"));
									DBHelper* dbReq = new DBHelper();
									ResultSet* ri = dbReq->createRequestItem(iRequestID, iCtr, iRet, esList->getInt("EdgeServerID"));
									int iRequestItemID;
									if (ri == NULL) {
										cout<<"error 3"<<endl;
										bError = true;
									} else {
										if (ri->rowsCount()>0) {
											ri->first();
											iRequestItemID = ri->getInt("RequestItemID");
											dbReq->disconnectDB();
										} else {
											cout<<"error 4"<<endl;
											bError = true;
										}
									}
									if (iRet < 0 && iRequestItemID > 0) {
										DBHelper* dbUpRIS = new DBHelper();
										dbUpRIS->updateRequestItemStatus(iRequestItemID, STATUS_ERROR); // 3 = Error
										dbUpRIS->disconnectDB();
										bError = true;
									}
									if (bError)
										break;
									if (!esList->next())
										esList->first();
								} catch (exception ex) {
									cout<<"chunkargs error "<<ex.what()<<endl;
								}
							}
							if (bError) {
								DBHelper* dbUpRS = new DBHelper();
								dbUpRS->updateRequestStatus(iRequestID, STATUS_ERROR);
								dbUpRS->disconnectDB();
							}
						} else {
							cout<<"error 5"<<endl;
							bError = true;
						}
					}
				} else {
					cout<<"error 6"<<endl;
					bError = true;
				}
			} else {
				cout<<"error 7"<<endl;
				bError = true;
			}
		}

		//delete esListl;
		//delete creq;
		dbES->disconnectDB();

		if (bError) {
			logger->logMessage("NULL", "NULL", "NULL", "MasterServer::ProcessFile", "Error in processing chunk", fileName);
			return -3; // es error
		}

	} catch(exception ex) {
		ExceptionHandler::processException(&ex, "MasterServer::ProcessFile", "Caught exception");
		return -1;
	} catch(...) {
		ExceptionHandler::processException("MasterServer::ProcessFile", "Caught exception");
		return -1;
	}
}



int MasterServer::ProcessVideoFile(int VendorID, char* fileName)
{

	try {
		DBHelper* dbChReq = new DBHelper();
		Logger* logger = new Logger();
		CommonUtilities* utilities = new CommonUtilities();
		CloudClient* client = new CloudClient((char*)HDFS_CLOUD_SERVER_HOST, HDFS_CLOUD_SERVER_PORT);
		CDNBitStreamExtract *cdnBitExtract=new CDNBitStreamExtract();
		hdfsFileInfo* fileInfo;
		time_t fiLastModified;
		size_t fiFileSize;

		bool bError = false;

		fileInfo = client->getFileInfoFromHDFS(fileName);
		fiLastModified = fileInfo->mLastMod;
		fiFileSize = fileInfo->mSize;

		// check this file in request table
		ResultSet* req = dbChReq->checkRequest(VendorID, fileName);
		if (req == NULL) {
			bError = true;
		} else {
			if (req->rowsCount() > 0) {
				req->first();
				int iOldRequestID = req->getInt("RequestID");
				SQLString sqls = req->getString("DateUpdated");
				char* OldDateUpdated = (char*)malloc(sqls.length());
				memcpy((void*)OldDateUpdated,(void*)sqls.c_str(),sqls.length());
				size_t OldSize = (size_t)req->getInt64("FileSize");
				dbChReq->disconnectDB();
				cout<<"iOldRequestID = "<<iOldRequestID<<" OldDateUpdated = "<<OldDateUpdated<<" OldSize = "<<OldSize<<endl;
				if (fileInfo != NULL) {
					if (utilities->isUpdated(fiLastModified, fiFileSize, OldDateUpdated, OldSize))
					{
						//BLOCK TO DELete inactive Request folders

						DBHelper* dbARI = new DBHelper();
						ResultSet* ari = dbARI->getActiveRequestItems(VendorID, fileName);

						if (ari == NULL)
						{
							bError = true;
						}
						else
						{
							if (ari->rowsCount() > 0)
							{
								cerr<<"Entered the ari block"<<endl;
								while(ari->next())
								{

									//cerr<<"chunkindex is :"<<ari->getInt("ChunkIndex")<<endl;
									//cerr<<"host name is :"<<(char*)ari->getString("HostName").c_str();
									//cerr<<"PORT name is :"<<(char*)ari->getString("ServicePort").c_str();

									// TODO
									// call method to delete chunk on edge server
									client->deleteFileFolderfromEdgeServer(VendorID, fileName,ari->getInt("ChunkIndex"), (char*)ari->getString("HostName").c_str(), ari->getInt("ServicePort"));

								}
							} else {
								bError = true;
							}
						}

						delete ari;




						DBHelper* dbMkReq = new DBHelper();
						dbMkReq->makeRequestInactive(iOldRequestID);
						dbMkReq->disconnectDB();
					} else {
						return 1;
					}
				} else {
					bError = true;
				}
			}
		}

		delete req;

		if (bError) {
			logger->logMessage("NULL", "NULL", "NULL", "MasterServer::ProcessFile", "Error in validating current Request queue", fileName);
			return -2; // db error
		}

		DBHelper* dbESG = new DBHelper();
		int iEdgeServerGroupID;
		ResultSet* esg = dbESG->getEdgeServerGroup();
		cout<<"here 1"<<endl;
		if (esg == NULL) {
			cout<<"here 2"<<endl;
			bError = true;
		} else {
			if (esg->first()) {
				cout<<"here 3"<<endl;
				iEdgeServerGroupID = esg->getInt("EdgeServerGroupID");
			} else {
				cout<<"here 4"<<endl;
				bError = true;
			}
		}

		delete esg;
		dbESG->disconnectDB();

		if (bError) {
			logger->logMessage("NULL", "NULL", "NULL", "MasterServer::ProcessFile", "Error in getting Edge Server Group", fileName);
			return -3; // es error
		}

		DBHelper* dbES = new DBHelper();
		ResultSet* esList = dbES->getEdgeServersInGroup(iEdgeServerGroupID);
		int iChunkCount = 0;
		int iRequestID;
		cout<<"here 5"<<endl;
		if (esList == NULL)
		{
			cout<<"error 1"<<endl;
			bError = true;
		}
		else
		{
			if (esList->rowsCount() > 0)
			{
				esList->first();

				//Invoke CDN Bit Extractor method to select the layers to be transferred
				cdnBitExtract->SelectLayerIdstoExtract(fileName);

				//iChunkCount = client->calculateNumberOfChunks(fileName);
				cout<<"iNumLayersMinus1 = "<<cdnBitExtract->iNumLayersMinus1<<endl;
				if ((cdnBitExtract->iNumLayersMinus1) > 0)
				{
					cout<<"here 6"<<endl;
					char* stime = new char[22];
					utilities->getTimeString(fiLastModified, stime, 22);
					cout<<"fiLastModified = "<<fiLastModified<<endl;
					cout<<"fiFileSize = "<<fiFileSize<<endl;
					cout<<stime<<endl;
					DBHelper* dbCReq = new DBHelper();
					ResultSet* creq = dbCReq->createRequest(VendorID, fileName, stime, fiFileSize, cdnBitExtract->iNumLayersMinus1, iEdgeServerGroupID);
					cout<<"here 7"<<endl;
					if (creq == NULL)
					{
						cout<<"error 2"<<endl;
						bError = true;
					} else
					{
						cout<<"here 8"<<endl;
						if (creq->rowsCount() > 0) {
							creq->first();
							iRequestID = creq->getInt("RequestID");
							cout<<"iRequestID ="<<iRequestID<<endl;
							dbCReq->disconnectDB();
							//for(int iCtr = 1; iCtr <= iChunkCount; iCtr++)
							for (int iCtr = 0; iCtr<=cdnBitExtract->iNumLayersMinus1; iCtr++)
							{
								try
								{
									/*
									struct chunkargs* args = (struct chunkargs*)malloc(sizeof(struct chunkargs));
									cout<<"host name = "<<(char*)esList->getString("HostName").c_str()<<endl;
									cout<<"host IP = "<<(char*)esList->getString("HostIP").c_str()<<endl;
									args->edgeserver_host = (char*)esList->getString("HostIP").c_str();
									args->edgeserver_port = esList->getInt("ServicePort");
									args->VendorID = VendorID;
									args->hdfsfileName = fileName;
									args->chunkIndex = iCtr;
									args->totalChunks = iChunkCount;
									*/
									//int iRet = client->processChunk(VendorID, fileName, iCtr, iChunkCount, (char*)esList->getString("HostName").c_str(), esList->getInt("ServicePort"));

									// Take CAre of teh Return type
									int iRet=cdnBitExtract->processLayer(VendorID,fileName,cdnBitExtract->iLayerIds[iCtr],cdnBitExtract->iNumLayersMinus1,(char*)esList->getString("HostName").c_str(), esList->getInt("ServicePort"));

									DBHelper* dbReq = new DBHelper();
									ResultSet* ri = dbReq->createRequestItem(iRequestID, cdnBitExtract->iLayerIds[iCtr], iRet, esList->getInt("EdgeServerID"));
									int iRequestItemID;
									if (ri == NULL)
									{
										cout<<"error 3"<<endl;
										bError = true;
									} else
									{
										if (ri->rowsCount()>0)
										{
											ri->first();
											iRequestItemID = ri->getInt("RequestItemID");
											dbReq->disconnectDB();
										} else
										{
											cout<<"error 4"<<endl;
											bError = true;
										}
									}
									if (iRet < 0 && iRequestItemID > 0) {
										DBHelper* dbUpRIS = new DBHelper();
										dbUpRIS->updateRequestItemStatus(iRequestItemID, STATUS_ERROR); // 3 = Error
										dbUpRIS->disconnectDB();
										bError = true;
									}
									if (bError)
										break;
									if (!esList->next())
										esList->first();
								} catch (exception ex)
								{
									cout<<"Layer transfer error "<<ex.what()<<endl;
								}
							}
							if (bError) {
								DBHelper* dbUpRS = new DBHelper();
								dbUpRS->updateRequestStatus(iRequestID, STATUS_ERROR);
								dbUpRS->disconnectDB();
							}
						} else {
							cout<<"error 5"<<endl;
							bError = true;
						}
					}
				} else {
					cout<<"error 6"<<endl;
					bError = true;
				}
			} else {
				cout<<"error 7"<<endl;
				bError = true;
			}
		}

		//delete esListl;
		//delete creq;
		dbES->disconnectDB();

		if (bError) {
			logger->logMessage("NULL", "NULL", "NULL", "MasterServer::ProcessFile", "Error in processing chunk", fileName);
			return -3; // es error
		}

		return 10;

	}
	catch(exception ex)
	{
		ExceptionHandler::processException(&ex, "MasterServer::ProcessVideoFile", "Caught exception");
		return -1;
	}
	catch(...) {
		ExceptionHandler::processException("MasterServer::ProcessVideoFile", "Caught exception");
		return -1;
	}
}

